package com.ctyun.dwi

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import com.shujia.utils.SparkTool

object DwiResRegnMergelocationMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    // 导入隐式转换及函数
    import spark.implicits._
    import org.apache.spark.sql.functions._
    // 在Spark SQL中使用自定义函数


    // 读取hive中ods层中的oidd的数据
    val oidd: DataFrame = spark.table("ods.ods_oidd")

    // 将开始时间、结束时间分成两列
    oidd
      // 取对应分区的数据
      .where($"day_id" === day_id)
      // 去除重复数据
      //      .select()
      //      .distinct()
      // withColumn可以给数据增加一列
      .withColumn("start_t", split($"start_time", ",")(1)) // 提取业务的开始时间
      .withColumn("end_t", split($"start_time", ",")(0)) // 提取业务的结束时间


      // 在时间轴上进行聚类 主要解决同一个网格内部相邻两条数据有时间交叠的问题
      // 按用户”分组“ 再按 开始时间 排序 获取上一条数据的网格编号
      .withColumn("last_grid", lag("grid_id", 1) over Window.partitionBy("mdn").orderBy("start_t"))
      // 用当前数据的grid_id 进行对比 上一条数据的grid_id 如果相同则置0，不同则置1
      .withColumn("flag", when($"grid_id" === $"last_grid", 0).otherwise(1))
      // 对flag列进行累计求和
      .withColumn("grp", sum($"flag") over Window.partitionBy("mdn").orderBy("start_t"))
      // 按照grp进行分组 取每个组内的 时间的最小值及最大值
      .groupBy("mdn", "county_id", "longi", "lati", "bsid", "grid_id", "grp")
      .agg(min("start_t") as "start_t", max("end_t") as "end_t")


      // 基于开始时间排序，取每一条数据的前一条数据的经纬度 作为新的一列 lag
      .withColumn("last_lg", lag($"longi", 1) over Window.partitionBy($"mdn").orderBy($"start_t")) // 取上一条数据的经度
      .withColumn("last_lat", lag($"lati", 1) over Window.partitionBy($"mdn").orderBy($"start_t")) // 取上一条数据的纬度

      // 由取上一条数据的结束时间end_t 变为 取上一条数据的 start_t 主要为了解决不同网格之间相邻两条数据有时间交叠的问题
      .withColumn("last_end_time", lag($"start_t", 1) over Window.partitionBy($"mdn").orderBy($"start_t")) // 取上一条数据的结束时间

      // 计算相邻两条位置记录之间的时间间隔
      .withColumn("diff_time", unix_timestamp($"start_t", "yyyyMMddHHmmss") - unix_timestamp($"last_end_time", "yyyyMMddHHmmss"))
      // 基于经纬度计算距离
      .withColumn("distance", when($"last_lg".isNull, 1).otherwise(calculateLength($"longi", $"lati", $"last_lg", $"last_lat")))
      // 根据距离及时间间隔计算速度
      .withColumn("speed", round($"distance" / $"diff_time", 3))

      // 将“超速”的数据进行过滤
      .where($"speed" <= 340)

      // 整理数据 将需要的字段取出来
      .select(
        // 对敏感数据进行脱敏操作
        upper(md5(concat($"mdn", expr("'shujia'"))) as "mdn"),
        $"start_t" as "start_date",
        $"end_t" as "end_date",
        $"county_id",
        $"longi",
        $"lati",
        $"bsid",
        $"grid_id")

      // 将结果保存到文件
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dwi/dwi_res_regn_mergelocation_msk_d/day_id=$day_id")


    // 增加分区
    spark.sql(
      s"""
         |alter table dwi.dwi_res_regn_mergelocation_msk_d  add if not exists partition(day_id='$day_id')
         |""".stripMargin)
  }

  /**
   * 1、使用maven将代码打成jar包并上传
   * 2、如果开启了Spark的historySever服务需要给/user/spark/applicationHistory目录通过acl设置权限
   * hdfs dfs -setfacl -R -m user:dwi:rwx /user/spark/applicationHistory
   * 3、使用spark-submit提交代码：
   * spark-submit --master yarn-client --class com.ctyun.dwi.DwiResRegnMergelocationMskDay --jars common-1.0.jar dwi-1.0.jar 20220527
   * 4、查看结果目录的文件大小
   * hdfs dfs -du -h /daas/motl/dwi/dwi_res_regn_mergelocation_msk_d/
   */

}
